package de.pyranja.storage;

import static com.google.common.base.Preconditions.checkNotNull;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.HashMultiset;
import com.google.common.collect.Multiset;
import com.google.common.hash.HashCode;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import com.google.common.io.ByteStreams;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;

import de.pyranja.storage.exception.BinaryStorageException;
import de.pyranja.storage.exception.BlobNotFoundException;

/**
 * An in-memory mock implementation of binary storage. Intended for prototyping,
 * testing and evaluation of design. This implementation is not thread-safe and
 * does not persist binary data.
 * @author Chris Borckholder
 */
public class InMemoryBinaryStorage implements BinaryStorage {
	/* slf4j-logger */
	private final static Logger log = LoggerFactory.getLogger(InMemoryBinaryStorage.class);
	// hasher
	private static final HashFunction HASH_ALGORITHM = Hashing.md5(); 
	
	// holds data
	private Map<String,byte[]> storage;
	// holds references
	private Multiset<String> references;
	
	public InMemoryBinaryStorage() {
		super();
		storage = new HashMap<>();
		references = HashMultiset.<String>create();
	}

	@Override
	public Blob store(InputStream in) {
		checkNotNull(in);
		try {
			byte[] data = ByteStreams.toByteArray(in);
			HashCode hash = HASH_ALGORITHM.hashBytes(data);
			String id = hash.toString();
			storage.put(id,data);
			references.add(id);
			return new Blob(id, data.length);
		} catch (IOException e) {
			throw new BinaryStorageException("error on reading data",e);
		} finally {
			try {
				in.close();
			} catch (IOException e) {
				log.error("failed to close stream",e);
			}
		}
	}

	@Override
	public InputStream fetch(String id) {
		checkNotNull(id);
		byte[] data = storage.get(id);
		if (data != null) {
			return new ByteArrayInputStream(data);
		} else {
			throw new BlobNotFoundException(id);
		}
	}

	@Override
	public ListenableFuture<String> claim(String id) {
		checkNotNull(id);
		if (references.contains(id)) {
			references.add(id);
			SettableFuture<String> response = SettableFuture.<String>create();
			response.set(id);
			return response;
		} else {
			throw new BlobNotFoundException(id);
		}
	}

	@Override
	public ListenableFuture<String> release(String id) {
		checkNotNull(id);
		references.remove(id);
		if (references.count(id) < 1) {
			storage.remove(id);
		}
		SettableFuture<String> response = SettableFuture.<String>create();
		response.set(id);
		return response;
	}
}
